/*-----------------------------------------------------------------------------
 * 
 *
 *-----------------------------------------------------------------------------
 */

#include <pthread.h>
#include "common/include/stock_common.h"
#include "common/include/stock_update.h"
#include "common/include/stock_timesync.h"
#include "server/include/stock_server.h"
#include "server/include/stock_server_queue.h"
#include "server/include/stock_server_stream.h"
#include "server/database/include/stock_database.h"


/*
 * Stream server statistics
 */
stock_stream_stats_t stock_server_stream_stats;


/*
 * Database of active and recently disconnected clients
 */
avl_tree *stock_server_stream_clients;


/*
 * Pool of threads from which stream handler threads are spawned - one for each
 * new client that connects to the server. 
 */
static pthread_t stock_stream_threadpool[STOCK_MAX_STREAM_CLIENTS];


/*
 * Find a free thread in the threadpool
 */
static pthread_t *
stock_threadpool_find(void) 
{
    int index;

    for (index = 0; index < STOCK_MAX_STREAM_CLIENTS; index++) {
        if (stock_stream_threadpool[index] == 0UL) {
            return &stock_stream_threadpool[index];
        }
    }

    return NULL;
}


/*
 * Return a thread to the threadpool
 */
static void
stock_threadpool_return(pthread_t tid)
{
    int index;

    for (index = 0; index < STOCK_MAX_STREAM_CLIENTS; index++) {
        if (stock_stream_threadpool[index] == tid) {
            stock_stream_threadpool[index] = 0UL;
            break;
        }
    }

    return;
}


/*
 * Main function that is run by spawned threads for clients.  Recv's updates
 * and queue's it to the server queue
 */
static void *
stock_stream_handler(void *arg)
{
    stock_update update;
    int          rc; 
    int          client_sock = *(int *) arg;

    while (1) {

        rc = stock_update_recv(client_sock, &update);
        
        if (rc == 0) {

            rc = stock_database_filter(&update);

            if (rc != 0) {
                /*
                 * If the database filter drops the update because of unsynced
                 * time, then send a timesync packet to the client.
                 */
                if (rc == STOCK_DATABASE_FILTER_UNSYNC) {
                    stock_timesync_send(client_sock, &update);  
                }

                continue;
            }

            stock_server_enqueue(&update);
                         
        } else if (rc == STOCK_ECONNRESET) {
            /*
             * Client has closed the connection. Break out of the read loop and
             * terminate this thread
             */
            stock_server_stream_stats.client_closed++;
            break;
        } else if (rc == STOCK_EMALFORMED) {
            // TODO: stats
            continue;
        } else if (rc == STOCK_EMSGSIZE) {
            /*
             * We should *never* recieve updates with bad sizes since we are
             * using the MSG_WAITALL flag when we recv data from the socket.
             * Calls to socket_update_recv should always pick up full update 
             * messages and reading when a partial update is sitting in the 
             * socket buffer should be blocking till the full update arrives
             *
             * This should be considered a Fatal Error in the stream and the
             * connection should be reset
             */
            error("Bad update size. Resetting connection");
            // TODO: stats
            break;
        }
    }

    stock_threadpool_return(pthread_self());

    return NULL;
}


/*
 * Initialize the stream based server
 */
static void 
stock_server_stream_init()
{
    /*
     * Initialize the threadpool array
     */
    memset(stock_stream_threadpool, 0, 
           sizeof(pthread_t) * STOCK_MAX_STREAM_CLIENTS);

    /*
     * Initialize the stats
     */
    memset(&stock_server_stream_stats, 0, sizeof(stock_stream_stats_t));
}


/* 
 * Stream server accept loop
 */
void
stock_server_stream(int server_sock)
{
    int                 client_sock = -1;
    struct sockaddr_in  client_addr;
    socklen_t           client_len;
    pthread_t          *thread;

    stock_server_stream_init();

    /*
     * Remaining setup for the socket
     */
    listen(server_sock, STOCK_MAX_STREAM_CLIENTS);

    while (1) {
        client_sock = accept(server_sock, (struct sockaddr *) &client_addr,
                             &client_len);

        if (client_sock < 0) {  
            stock_server_stream_stats.accept_failed++;
            error("accept() call failed");
            continue;
        } else {
            printf("new client\n");
            stock_server_stream_stats.accepted++;
        }
    
        thread = stock_threadpool_find();

        if (thread == NULL) {
            stock_server_stream_stats.rejected++;
            warning("stream threadpool exhausted - ignoring new connection");
            continue;
        }

        pthread_create(thread, NULL, stock_stream_handler, &client_sock);
    }
}

